package com.chief.tableApi;


import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.ConnectTableDescriptor;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;

import static org.apache.flink.table.api.Expressions.$;

public class TestTableFromFile {


    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setParallelism(1);

        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(streamExecutionEnvironment);

        String path = "E:\\bigdata-start\\005 flink\\flinkCode\\src\\main\\java\\com\\chief\\window\\data.txt";
        tableEnvironment.executeSql("create table test(\n" +
                "name string,ctime string,score integer\n" +
                ") WITH ('connector'='filesystem','path'='" + path + "','format'='csv')");
        Table select = tableEnvironment.from("test").select($("name"));
        tableEnvironment.toDataStream(select).print();

        TableResult tableResult = tableEnvironment.executeSql("select name,score from test");
//        tableResult.print();
        CloseableIterator<Row> collect = tableResult.collect();
        System.out.println("===============");
        while (collect.hasNext()) {
            System.out.println(collect.next().getField("name"));
        }
        System.out.println("2222222222222222222");


        tableEnvironment.connect(new FileSystem().path(path))
                .withFormat(new Csv())
                .withSchema(new Schema().field("name", DataTypes.STRING())
                        .field("ctime",DataTypes.STRING()).field("score",DataTypes.INT()))
                .createTemporaryTable("test2");

        Table test2 = tableEnvironment.from("test2");
        test2.printSchema();
        select =test2 .select($("name"),$("ctime"));
        tableEnvironment.toDataStream(select).print();

        streamExecutionEnvironment.execute();
    }

}
